/**
 * 
 */
package stc.skymobi.distributed;

import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author hp
 *
 */
public class GlobalMutex {
	
	public static interface LockCallback {
		
		public	void 	lockAcquired(String lockedId);
		
		public 	void	lockBlocked(String lockedId);
		
		public	void	exceptionCaught(Throwable e);
	}
	
    private static final Logger logger = LoggerFactory.getLogger(GlobalMutex.class);
    private final String LOCK_PREFIX = "lock-";
    
	private	ZooKeeper	zk;
	private	String		root;
	
	private	boolean isZkConnected() {
		return (null != zk ) && ( zk.getState() == ZooKeeper.States.CONNECTED );
	}
	
	/**
	 * @param zk the zk to set
	 */
	public void setZk(ZooKeeper zk) {
		this.zk = zk;
	}

	public	GlobalMutex(String lockRoot) {
		this.root = lockRoot;
	}
	
	public 	void	start() throws KeeperException, InterruptedException {
		if ( !isZkConnected() ) {
			throw new RuntimeException("ZooKeeper not connected.");
		}
		
		if ( zk.exists(root, false) == null ) {
			if ( logger.isInfoEnabled() ) {
				logger.info("Mutex root [" + root + "] does not exists, creating");
			}
			
			try {
				zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			catch (NodeExistsException e) {
            	//	other client create this path first
				if ( logger.isWarnEnabled() ) {
					logger.warn("Mutex root [" + root + "] exists, just skip.");
				}
			}
		}
	}

	private String createLockZNode(String lockPath) throws KeeperException, InterruptedException {
		String lockZNode = null;
	
        try
        {
            lockZNode = zk.create(lockPath  + "/"  + LOCK_PREFIX, 
            		new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        }
        catch (NoNodeException e)
        {
        	if ( logger.isInfoEnabled() ) {
        		logger.info("lockPath [" + lockPath + "] does not exist, creating");
        	}
        	
            try {
            	zk.create(lockPath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            catch (NodeExistsException existException) {
            	//	other client create this path first
            	if ( logger.isWarnEnabled() ) {
            		logger.warn("lockPath [" + lockPath + "] exists, just skip.");
            	}
			}
            
            lockZNode = zk.create(lockPath + "/" + LOCK_PREFIX, 
            		new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        }

        return lockZNode;
    }

	private void checkLockStatus(final String lockPath, final String lockZNode, final LockCallback callback) {
		
		try {
			while (true) {
				// check what is our ID (sequence number at the end of file name
				// added by ZK)
				int mySeqNum = Integer.parseInt(lockZNode.substring(lockZNode
						.lastIndexOf('-') + 1));
				int previousSeqNum = -1;
				String predessor = null;

				// get all children of lock znode and find the one that is just
				// before us, if
				// any. This must be inside loop, as children might get deleted
				// out of order because
				// of client disconnects. We cannot assume that the file that is
				// in front of us this
				// time, is there next time. It might have been deleted even
				// though earlier files
				// are still there.
				List<String> children = zk.getChildren(lockPath, false);
				if (children.isEmpty()) {
					if (logger.isErrorEnabled()) {
						logger.error("No children in ["
										+ lockPath
										+ "] although one was just created. just failed lock progress.");
					}
					callback.exceptionCaught(new RuntimeException("previous created [" + lockZNode + "] not exists."));
					return;
				}

				for (String child : children) {
					if (logger.isTraceEnabled()) {
						logger.trace("child: " + child);
					}

					int otherSeqNum = Integer.parseInt(child.substring(child
							.lastIndexOf('-') + 1));
					if ((otherSeqNum < mySeqNum)
							&& (otherSeqNum > previousSeqNum)) {
						previousSeqNum = otherSeqNum;
						predessor = child;
					}
				}

				// our sequence number is smallest, we have the lock
				if (-1 == previousSeqNum) {
					if (logger.isTraceEnabled()) {
						logger.trace("No smaller znode sequences, " + lockZNode
								+ " acquired lock");
					}

					callback.lockAcquired(lockZNode);
					return;
				}

				Watcher watcher = new Watcher() {

					public void process(WatchedEvent event) {
						checkLockStatus(lockPath, lockZNode, callback);
					}
				};

				if (zk.exists(lockPath + "/" + predessor, watcher) == null) {
					if (logger.isTraceEnabled()) {
						logger.trace(predessor + " does not exists, "
								+ lockZNode + " acquired lock");
					}

					callback.lockAcquired(lockZNode);
					return;
				} else {
					if (logger.isTraceEnabled()) {
						logger.trace(predessor + " is still here, " + lockZNode
								+ " must blocked for wait");
					}
					callback.lockBlocked(lockZNode);
					return;
				}
			}
		}
		catch (Throwable e) {
			callback.exceptionCaught(e);
		}
	}
	
	public void	acquireLock(String resName, LockCallback callback) {
		
		if ( !isZkConnected() ) {
			callback.exceptionCaught( new RuntimeException("ZooKeeper not connected.") );
		}
		
		String lockPath = root + "/" + resName;

		String	lockZNode = null;
		try {
			lockZNode = createLockZNode(lockPath);
			if (logger.isTraceEnabled()) {
				logger.trace("lockZNode created, [" + lockZNode + "]");
			}
		}
		catch (Throwable e) {
			callback.exceptionCaught(e);
		}
		
		checkLockStatus(lockPath, lockZNode, callback);
	}
	
	public void releaseLock(String lockedId) {
        if (logger.isTraceEnabled()) {
            logger.trace("deleting ["  + lockedId + "]");
        }
        
        try {
            zk.delete(lockedId, -1);
        }
        catch (Exception e) {
            // We do not do anything here. The idea is to check that everything goes OK when
            // locking and let unlock always succeed from client's point of view. Ephemeral
            // nodes should be taken care of by ZooKeeper, so ignoring any errors here should
            // not break anything.
        }
	}
}
